package com.mo.axion.handler.receiver.kafka;

import cn.hutool.core.text.StrPool;
import com.mo.axion.handler.utils.GroupIdMappingUtils;
import com.mo.axion.support.mq.constants.MessageQueuePipeline;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.kafka.annotation.KafkaListenerAnnotationBeanPostProcessor;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.lang.reflect.Method;
import java.util.List;

/**
 * @author Mo
 * @createTime 2024/2/17 18:10
 * @description 消费者启动类
 */
@Slf4j
@Service
@ConditionalOnProperty(name = "axion.mq.pipeline", havingValue = MessageQueuePipeline.KAFKA)
public class ReceiverStart {
    /**
     * receiver的消费方法常量
     */
    private static final String RECEIVER_METHOD_NAME = "Receiver.consumer";

    /**
     * 获取所有消费者类型的groupId
     */
    private static List<String> groupIdList = GroupIdMappingUtils.getAllGroupIdList();

    private static Integer index = 0;
    @Autowired
    private ApplicationContext context;

    @Autowired
    private ConsumerFactory consumerFactory;

    /**
     * 给每个Receiver对象的consumer方法 @KafkaListener赋值相应的groupId
     * @return
     */
    @Bean
    public static KafkaListenerAnnotationBeanPostProcessor.AnnotationEnhancer groupIdEnhancer() {
        return (attributes, element) -> {
            if (element instanceof Method) {
                String name = ((Method) element).getDeclaringClass().getSimpleName() + StrPool.DOT + ((Method) element).getName();
                if (RECEIVER_METHOD_NAME.equals(name)) {
                    attributes.put("groupId", groupIdList.get(index++));
                }
            }
            return attributes;
        };
    }


    /**
     * 为每个渠道不同的消息类型 创建一个Receiver对象
     */
    @PostConstruct
    public void init() {
        for (int i = 0; i < groupIdList.size(); i++) {
            context.getBean(Receiver.class);
        }
    }

}
